/*
 * Copyright 2021 DataCanvas
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.dingodb.calcite.executor;

import io.dingodb.calcite.grammar.dql.SqlShowColumns;
import io.dingodb.common.util.SqlLikeUtils;
import io.dingodb.common.util.Utils;
import io.dingodb.meta.DdlService;
import io.dingodb.meta.entity.Column;
import io.dingodb.meta.entity.IndexTable;
import io.dingodb.meta.entity.InfoSchema;
import io.dingodb.meta.entity.Table;
import lombok.Setter;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.parser.SqlParserUtil;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class ShowColumnsExecutor extends QueryExecutor {

    @Setter
    public SqlNode sqlNode;

    private final String schemaName;
    private final String tableName;

    private final String sqlLikePattern;

    public ShowColumnsExecutor(SqlNode sqlNode) {
        SqlShowColumns showColumns = (SqlShowColumns) sqlNode;
        this.schemaName = showColumns.schemaName;
        this.tableName = showColumns.tableName;
        this.sqlLikePattern = showColumns.sqlLikePattern;
    }

    @Override
    public Iterator<Object[]> getIterator() {
        List<Object[]> tuples = new ArrayList<>();
        List<List<String>> columnList = getColumnFields();
        for (List<String> values : columnList) {
            Object[] tuple = values.toArray();
            tuples.add(tuple);
        }
        return tuples.iterator();
    }

    @Override
    public List<String> columns() {
        List<String> columns = new ArrayList<>();
        columns.add("Field");
        columns.add("Type");
        columns.add("Null");
        columns.add("Key");
        columns.add("Default");
        return columns;
    }

    private List<List<String>> getColumnFields() {
        InfoSchema is = DdlService.root().getIsLatest();
        Table table = is.getTable(schemaName, tableName);
        if (table == null) {
            throw new RuntimeException("Table " + tableName + " doesn't exist");
        }
        List<IndexTable> uniqueIndexList = table.getIndexes().stream()
            .filter(indexTable -> indexTable.isUnique() && indexTable.visible).toList();

        List<Column> columns = table.getColumns();
        List<List<String>> columnList = new ArrayList<>();
        boolean haveLike = !StringUtils.isBlank(sqlLikePattern);
        for (Column column : columns) {
            if (column.getState() != 1) {
                continue;
            }
            List<String> columnValues = new ArrayList<>();

            String columnName = column.getName();
            if (haveLike && !SqlLikeUtils.like(columnName, sqlLikePattern)) {
                continue;
            }

            columnValues.add(columnName);
            String type = column.getSqlTypeName();
            if (type.equalsIgnoreCase("INTEGER")) {
                type = "int";
                if (column.getPrecision() > 0) {
                    type = type + "(" + column.getPrecision() + ")";
                }
            } else if (type.equals("VARCHAR") || type.equals("CHAR")) {
                if (column.getPrecision() > 0) {
                    type = type + "(" + column.getPrecision() + ")";
                }
            } else if (type.equalsIgnoreCase("DECIMAL")) {
                if (column.getPrecision() > 0 && column.getScale() >= 0) {
                    type = type + "(" + column.getPrecision() + "," + column.getScale() + ")";
                }
            }
            columnValues.add(type.toLowerCase());
            columnValues.add(column.isNullable() ? "YES" : "NO");
            String key = " ";
            if (column.isPrimary()) {
                key = "PRI";
            } else {
                boolean uniqueKey = uniqueIndexList.stream().anyMatch(indexTable -> {
                    return indexTable.getColumns().stream()
                        .anyMatch(column1 -> column1.getName().equalsIgnoreCase(column.getName()));
                });
                if (uniqueKey) {
                    key = "UNI";
                }
            }
            columnValues.add(key);
            String defaultValExpr;
            if ("VARCHAR".equalsIgnoreCase(column.getSqlTypeName())
                || "CHAR".equalsIgnoreCase(column.getSqlTypeName())) {
                defaultValExpr = Utils.decodePostgresUnicode(column.defaultValueExpr);
                if (defaultValExpr != null && defaultValExpr.startsWith("'") && defaultValExpr.endsWith("'")) {
                    defaultValExpr = SqlParserUtil.trim(defaultValExpr, "'");
                }
            } else {
                defaultValExpr = column.defaultValueExpr;
            }
            columnValues.add(defaultValExpr != null ? defaultValExpr : "NULL");

            columnList.add(columnValues);
        }
        return columnList;
    }
}
